package cn.demo.mq.pubsub.direct;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;

/**
 * @author Mr.Still
 * @version 1.0.0
 * @date 2021-10-18 15:25
 **/
public class Receiver02 {

    public static String QUEUE_NAME = "ps_error";

    //交换机
    public static String EXCHANGE_NAME = "exchange_direct";

    public static void main(String[] args) {
        Connection connection = null;
        Channel channel = null;
        try {
            //1，连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("8.129.129.84");
            factory.setPort(5672);
            factory.setVirtualHost("/demo");
            factory.setUsername("user");
            factory.setPassword("password");
            //2，创建连接对象
            connection = factory.newConnection();
            //3，创建通道
            channel = connection.createChannel();
            //每次接受消息《=1
            channel.basicQos(1);

            //4，指定接收消息的队列
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            //5，将指定队列绑定到交换机    routingKey or bindingKey = info,warning
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"qaq");
//            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"warning");

            //6，消息接收
            final Channel finalChannel = channel;
            Consumer consumer = new DefaultConsumer(finalChannel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    super.handleDelivery(consumerTag, envelope, properties, body);
                    String message = new String(body, StandardCharsets.UTF_8);
                    System.out.println("error ->>>>>>>>"+message);
                    try {
                        Thread.sleep(200);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //手动回执
                    finalChannel.basicAck(envelope.getDeliveryTag(),false);
                }
            };

            //队列名，true 消息收到进行删除 ，consumer
            //监听指定队列
            channel.basicConsume(QUEUE_NAME,false,consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}
